package priv.abadstring.study.haoke.im.websocket;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;
import priv.abadstring.study.haoke.im.bean.Message;
import priv.abadstring.study.haoke.im.dao.MessageDAO;
import priv.abadstring.study.haoke.im.mock.UserMockData;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * MessageHandler
 *
 * @author aBadString
 * @date 2021-07-30 17:53:50
 */
@Component
public class MessageHandler extends TextWebSocketHandler {

    @Autowired
    private MessageDAO messageDAO;
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    public static final Map<Long,WebSocketSession> SESSIONS = new ConcurrentHashMap<>();
    private static final ObjectMapper MAPPER = new ObjectMapper();

    /**
     * 建立连接后: 将当前用户的 session 放置到 map 中, 后面会使用相应的session通信
     */
    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        Long uid = (Long) session.getAttributes().get("uid");
        SESSIONS.put(uid, session);
    }

    /**
     * 收到 from 发送的消息
     */
    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage textMessage) {
        Long uid = (Long) session.getAttributes().get("uid");

        try {
            JsonNode jsonNode = MAPPER.readTree(textMessage.getPayload());
            Long toId = jsonNode.get("toId").asLong();
            String msg = jsonNode.get("msg").asText();

            Message message = Message.builder()
                    .from(UserMockData.USER_MAP.get(uid))
                    .to(UserMockData.USER_MAP.get(toId))
                    .msg(msg)
                    .build();
            message = this.messageDAO.saveMessage(message);
            String messageJson = MAPPER.writeValueAsString(message);
            System.out.println(messageJson);

            // 判断 to 用户是否在线
            WebSocketSession toSession = SESSIONS.get(toId);
            if(toSession != null && toSession.isOpen()){
                toSession.sendMessage(new TextMessage(messageJson));
                // 更新消息状态为已读
                this.messageDAO.updateMessageState(message.getId(), 2);
            }
            // 用户不在当前机器上, 则发送消息广播给其他节点实例去处理
            else {
                rocketMQTemplate.send(
                        // topic:TAG
                        "haoke-im:SEND_MESSAGE",
                        MessageBuilder.withPayload(messageJson).build()
                );
            }
        } catch (IOException | NullPointerException e ) {
            e.printStackTrace();
        }
    }
}
